package com.example.design.nioandnettty.practice.websocketLongCon;

import cn.hutool.core.collection.CollectionUtil;
import com.alibaba.fastjson.JSON;
import io.netty.channel.Channel;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import io.netty.handler.codec.http.websocketx.TextWebSocketFrame;
import lombok.extern.slf4j.Slf4j;
import org.springframework.util.StringUtils;

import java.util.Collection;
import java.util.Map;
import java.util.Optional;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;

@Slf4j
public class WebSocketHandler extends SimpleChannelInboundHandler<TextWebSocketFrame> {
    /**
     * 通道map，存储channel，用于群发消息，以及统计客户端的在线数量(如果是集群环境使用redis的hash数据类型存储即可)
     */
    private static Map<String, Channel> channelMap = new ConcurrentHashMap<>();
    /**
     * 任务map，存储future，用于停止队列任务
     */
    private static Map<String, Future> futureMap = new ConcurrentHashMap<>();
    /**
     * 存储channel的id和用户主键的映射，客户端保证用户主键传入的是唯一值(如果是集群中需要换成redis的hash数据类型存储即可)
     */
    private static Map<String, Long> clientMap = new ConcurrentHashMap<>();

    /**
     * 客户端发送给服务端的消息
     */
    @Override
    protected void channelRead0(ChannelHandlerContext ctx, TextWebSocketFrame msg) throws Exception {
        try {
            //接受客户端发送的消息
            MessageRequest messageRequest = JSON.parseObject(msg.text(), MessageRequest.class);

            // channelId
            String key = ctx.channel().id().asLongText();

            // 存储channel的id和用户的主键
            // clientMap的value可以根据业务改一下,比如：要推送设备的属性值给用户，但是不是所有的设备的属性值都推给用户，
            // 可以让客户端发送过来的消息里指定它需要哪些设备，只需要在MessageRequest在加一个List字段存储用户订阅的字段
            // 然后clientMap的value改为MessageRequest
            // 服务端给客户端推消息的时候，就可以根据MessageRequest知道，该推哪些设备的属性值给客户端了
            clientMap.put(key, messageRequest.getUnionId());

            // 处理客户端发来的消息
            log.info("接受客户端的消息......" + ctx.channel().remoteAddress() + "-参数[" + messageRequest.getUnionId() + "]");
        } catch (Exception e) {
            log.error("websocket服务器推送消息发生错误：", e);
        }
    }

    /**
     * 客户端连接时候的操作
     */
    @Override
    public void handlerAdded(ChannelHandlerContext ctx) throws Exception {
        log.info("一个客户端连接......" + ctx.channel().remoteAddress() + Thread.currentThread().getName());

        // 存储客户端和服务的通信的Chanel
        channelMap.put(ctx.channel().id().asLongText(), ctx.channel());

        // 周期性的发送心跳
        sendBeatToAllClient();
    }

    /**
     * 客户端掉线时的操作
     */
    @Override
    public void handlerRemoved(ChannelHandlerContext ctx) throws Exception {
        clear(ctx);
        log.info("一个客户端移除......" + ctx.channel().remoteAddress());
    }

    /**
     * 发生异常时执行的操作
     */
    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        clear(ctx);
        log.info("异常发生 " + cause.getMessage());
    }

    /**
     * 清理资源
     */
    private void clear(ChannelHandlerContext ctx) {
        String key = ctx.channel().id().asLongText();
        //移除通信过的channel
        channelMap.remove(key);
        //移除和用户绑定的channel
        clientMap.remove(key);
        //移除定时任务
        Optional.ofNullable(futureMap.get(key)).ifPresent(future -> {
            future.cancel(true);
            futureMap.remove(key);
        });
        //关闭长连接
        ctx.close();
    }

    /**
     * 发送消息给指定客户端
     */
    public static void sendMsgToClient(Long userId, String msg) {
        if (userId == null) {
            return;
        }

        if (CollectionUtil.isEmpty(clientMap)) {
            return;
        }

        for (String key : clientMap.keySet()) {
            Long user = clientMap.get(key);
            if (user.equals(userId)) {
                Channel channel = channelMap.get(key);
                if (channel == null) {
                    return;
                }
                channel.writeAndFlush(new TextWebSocketFrame(msg));
            }
        }
    }

    /**
     * 发送消息给所有客户端
     */
    public static void sendMsgToAllClient(String msg) {
        if (StringUtils.isEmpty(msg)) {
            return;
        }
        Collection<Channel> values = channelMap.values();
        if (CollectionUtil.isEmpty(values)) {
            return;
        }
        for (Channel channel : values) {
            channel.writeAndFlush(new TextWebSocketFrame(msg));
        }
    }

    /**
     * 统计在线客户数
     */
    public static Integer getOnlineClientNum() {
        if (CollectionUtil.isEmpty(channelMap)) {
            return 0;
        }
        return channelMap.size();
    }

    /**
     * 发送心跳
     */
    private void sendBeatToAllClient() {
        Collection<Channel> values = channelMap.values();
        for (Channel channel : values) {
            Future futured = futureMap.get(channel.id().asLongText());
            if (futured == null) {
                // 定时任务30秒发一次心跳
                Future future = channel.eventLoop().scheduleAtFixedRate(() -> channel.writeAndFlush(new TextWebSocketFrame("beat...")),
                        0, 30, TimeUnit.SECONDS);
                // 存储每个channel中的future，保证每个channel中只有一个定时任务在执行, 避免同一个channel有多个定时任务同时发心跳
                futureMap.put(channel.id().asLongText(), future);
            }
        }
    }

    public static Map<String, Channel> getChannelMap() {
        return channelMap;
    }

    public static Map<String, Future> getFutureMap() {
        return futureMap;
    }

    public static Map<String, Long> getClientMap() {
        return clientMap;
    }
}